package com.event.streaming.process


import java.util.Date
import scala.util.{Try, Success}
import org.apache.spark.sql.{SparkSession, DataFrame}
import org.apache.spark.sql.functions._

/* -----------------------------------------------------------------------------------------------------------

 * The following code is created by Wayne Shen for the Big Data Training
 *
 * Date: Sept. 29, 2018
 * Copyright @ iT21 Learning
 *
 ---------------------------------------------------------------------------------------------------------- */
@SerialVersionUID(20180722L)
class EventTransformer(spark: SparkSession) extends Serializable {
  //event start-time format
  protected val fmtEventStartTime = "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z"
  //invite time-stamp format
  protected val fmtInviteTimeStamp = "^\\d{4}-\\d{2}-\\d{2}\\s\\d{2}:\\d{2}:\\d{2}.*"
  //joined at time-stamp
  protected val fmtJoinTimeStamp = "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z"

  //the date-time format
  private val fmtDate = new java.text.SimpleDateFormat("yyyy-MM-dd")
  private val fmtDateTime = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
  //month format
  private val fmtMonth = new java.text.SimpleDateFormat("MM")
  //day of week format
  private val fmtDayOfWeek = new java.text.SimpleDateFormat("u")
  //hour format
  private val fmtHour = new java.text.SimpleDateFormat("HH")
  //year format
  private val fmtYear = new java.text.SimpleDateFormat("YYYY")

  //calculate the days
  private def calculateDays(dt: String, fmt: String): Option[Int] = {
    //check
    dt.matches(fmt) match {
      case true => Try((((new Date()).getTime() - fmtDate.parse(dt.substring(0, 10)).getTime())/(24*60*60*1000) + 1).toInt).toOption
      case _ => None
    }
  }

  //calculate the days from event start-time
  protected val daysFromStartTime = udf { (dt: String) => calculateDays(dt, fmtEventStartTime) }
  //calculate the start month, week-of-day, and hour
  protected val startFacts = udf {
    (dt: String) => {
      var month: Option[Int] = None
      var dayOfWeek: Option[Int] = None
      var hour: Option[Int] = None

      //check
      dt.matches(fmtEventStartTime) match {
        case true => {
          //the date
          Try(fmtDateTime.parse("%s %s".format(dt.substring(0, 10), dt.substring(11, 19)))) match {
            case Success(date) => {
              //organize the result
              month = Try(fmtMonth.format(date).toInt).toOption
              dayOfWeek = Try(fmtDayOfWeek.format(date).toInt).toOption
              hour = Try(fmtHour.format(date).toInt).toOption
            }
            case _ => { /* do nothing */ }
          }
        }
        case _ => { /* do notning */ }
      }
      Array(month, dayOfWeek, hour)
    }
  }
  //calculate the days from invite time-stamp
  protected val daysFromInviteTime = udf { (dt: String) => calculateDays(dt, fmtInviteTimeStamp) }
  //calculate the days from event start-time
  protected val daysFromJoinTime = udf { (dt: String) => calculateDays(dt, fmtJoinTimeStamp) }

  //check if locations are similar
  protected val locationSimilar = udf {
    (user_location: String, event_city: String, event_province: String, event_country: String) => if (user_location.contains(event_city) || user_location.contains(event_province) || user_location.contains(event_country)) 1 else 0
  }
  //calculate the percentage value
  protected val percentageValue = udf {
    (value: Double, min_value: Double, max_value: Double) => Try { if (value == null || min_value == null || max_value == null) 0.5D else (value - min_value) / (max_value - min_value) }.toOption.getOrElse(0.5D)
  }

  //event creator is friend
  protected def getEventCreatorIsFriend(dfUserEvent: DataFrame, tblUserFriend: String, tblUserFriendCount: String): DataFrame = {
    //register table
    dfUserEvent.createOrReplaceTempView("user_event")
    //sql statement
    val stmt: String = s"""SELECT
        ue.user_id,
        ue.event_id,
        ue.user_invited,
        ue.invite_ahead_days,
        ue.user_interested,
        ue.event_creator,
        ue.start_ahead_days,
        ue.event_start_month,
        ue.event_start_dayofweek,
        ue.event_start_hour,
        ue.event_city,
        ue.event_state,
        ue.event_country,
        ue.latitude,
        ue.longitude,
        CASE WHEN uf.friend_id IS NOT NULL THEN 1 ELSE 0 END AS event_creator_is_friend,
        CASE WHEN ufc.friend_count IS NOT NULL THEN ufc.friend_count ELSE 0 END AS user_friend_count
      FROM user_event ue
        LEFT JOIN $tblUserFriend uf ON ue.user_id = uf.user_id AND ue.event_creator = uf.friend_id
        LEFT JOIN $tblUserFriendCount ufc ON ue.user_id = ufc.user_id"""
    //execute
    spark.sql(stmt)

//    spark API
//    dfUserEvent.alias("ue").join(spark.sql("select * from $tblUserFriend").alias("uf"),col("ue.user_id") ===  col("uf.user_id") && col("ue.event_creator") === col("uf.friend_id"))
//      .select("ue.user_id")
  }

  //get the friend attend percentage
  protected def getFriendAttendPercentage(dfEventCreatorIsFriend: DataFrame, tblFriendAttendSummary: String): DataFrame = {
    //register
    dfEventCreatorIsFriend.createOrReplaceTempView("event_creator_is_friend")
    //statement
    val stmt: String = s"""SELECT
        ecif.user_id,
        ecif.event_id,
        ecif.invite_ahead_days,
        ecif.start_ahead_days,
        ecif.event_start_month,
        ecif.event_start_dayofweek,
        ecif.event_start_hour,
        ecif.event_city,
        ecif.event_state,
        ecif.event_country,
        ecif.latitude,
        ecif.longitude,
        ecif.event_creator_is_friend,
        ecif.user_friend_count,
        ecif.user_invited,
        ecif.user_interested,
        CASE WHEN fas.invited_friends_count IS NOT NULL THEN fas.invited_friends_count ELSE 0 END AS invited_friends_count,
        CASE WHEN fas.attended_friends_count IS NOT NULL THEN fas.attended_friends_count ELSE 0 END AS attended_friends_count,
        CASE WHEN fas.not_attended_friends_count IS NOT NULL THEN fas.not_attended_friends_count ELSE 0 END AS not_attended_friends_count,
        CASE WHEN fas.maybe_attended_friends_count IS NOT NULL THEN fas.maybe_attended_friends_count ELSE 0 END AS maybe_attended_friends_count,
        CASE WHEN ecif.user_friend_count != 0 AND fas.invited_friends_count IS NOT NULL THEN fas.invited_friends_count * 100 / ecif.user_friend_count ELSE 0 END as invited_friends_percentage,
        CASE WHEN ecif.user_friend_count != 0 AND fas.attended_friends_count IS NOT NULL THEN fas.attended_friends_count * 100 / ecif.user_friend_count ELSE 0 END as attended_friends_percentage,
        CASE WHEN ecif.user_friend_count != 0 AND fas.not_attended_friends_count IS NOT NULL THEN fas.not_attended_friends_count * 100 / ecif.user_friend_count ELSE 0 END as not_attended_friends_percentage,
        CASE WHEN ecif.user_friend_count != 0 AND fas.maybe_attended_friends_count IS NOT NULL THEN fas.maybe_attended_friends_count * 100 / ecif.user_friend_count ELSE 0 END as maybe_attended_friends_percentage
      FROM event_creator_is_friend ecif
        LEFT JOIN $tblFriendAttendSummary fas ON ecif.user_id = fas.user_id AND ecif.event_id = fas.event_id"""
    //execute
    spark.sql(stmt)
  }

  //get user-friend-event
  protected def getUserFriendEvent(dfFriendAttendPercentage: DataFrame, dfUserEventCount: DataFrame,
                                   tblUsers: String, tblEventAttendeeCount: String, tblUserAttendEventCount: String,
                                   tblEventCities: String, tblEventCountries: String, tblLocale: String, tblTimeZone: String): DataFrame = {
    //register
    dfFriendAttendPercentage.createOrReplaceTempView("friend_attend_percentage")
    dfUserEventCount.createOrReplaceTempView("user_event_count")
    //register udf
    spark.udf.register("age", (birth_year: Int) => fmtYear.format(new Date()).toInt - birth_year)
    spark.udf.register("daysFromJoinTime", daysFromJoinTime)
    spark.udf.register("locationSimilar", locationSimilar)
    //statement
    val stmt: String = s"""SELECT
        CAST(fap.user_interested AS int) AS user_interested,
        fap.user_id,
        fap.event_id,
        CASE WHEN l.locale_id IS NOT NULL THEN l.locale_id ELSE 0 END AS user_locale,
        CASE WHEN tz.time_zone_id IS NOT NULL THEN tz.time_zone_id ELSE 0 END AS user_time_zone,
        CASE WHEN u.gender = 'male' THEN 1 WHEN u.gender = 'female' THEN 0 ELSE -1 END AS user_gender,
        age(u.birth_year) AS user_age,
        daysFromJoinTime(u.joined_at) AS user_member_days,
        fap.user_friend_count,
        fap.invite_ahead_days AS user_invite_ahead_days,
        uec.event_count AS user_had_event_count,
        uaec.invited_count AS user_invited_event_count,
        uaec.attended_count AS user_attended_event_count,
        uaec.not_attended_count AS user_not_attended_event_count,
        uaec.maybe_attended_count AS user_maybe_attended_event_count,
        CAST(fap.user_invited AS int) AS user_invited,
        fap.invited_friends_count AS user_invited_friends_count,
        fap.attended_friends_count AS user_attended_friends_count,
        fap.not_attended_friends_count AS user_not_attended_friends_count,
        fap.maybe_attended_friends_count AS user_maybe_attended_friends_count,
        fap.invited_friends_percentage AS user_invited_friends_percentage,
        fap.attended_friends_percentage AS user_attended_friends_percentage,
        fap.not_attended_friends_percentage AS user_not_attended_friends_percentage,
        fap.maybe_attended_friends_percentage AS user_maybe_attended_friends_percentage,
        fap.start_ahead_days AS event_start_ahead_days,
        fap.event_start_month,
        CAST(fap.event_start_dayofweek AS int) AS event_start_dayofweek,
        fap.event_start_hour,
        iv.attend_count AS event_invited_user_count,
        yes.attend_count AS event_attended_user_count,
        no.attend_count AS event_not_attended_user_count,
        maybe.attend_count AS event_maybe_attended_user_count,
        ec.level AS event_city_level,
        et.level AS event_country_level,
        fap.event_creator_is_friend,
        locationSimilar(lower(u.location), lower(fap.event_city), lower(fap.event_state), lower(fap.event_country)) AS location_similar,
        fap.latitude,
        fap.longitude
        FROM friend_attend_percentage fap
          INNER JOIN $tblUsers u ON fap.user_id = u.user_id
          INNER JOIN user_event_count uec ON uec.user_id = fap.user_id
          INNER JOIN $tblEventAttendeeCount iv ON iv.event_id = fap.event_id AND iv.attend_type = 'invited'
          INNER JOIN $tblEventAttendeeCount yes ON yes.event_id = fap.event_id AND yes.attend_type = 'yes'
          INNER JOIN $tblEventAttendeeCount no ON no.event_id = fap.event_id AND no.attend_type = 'no'
          INNER JOIN $tblEventAttendeeCount maybe ON maybe.event_id = fap.event_id AND maybe.attend_type = 'maybe'
          LEFT JOIN $tblUserAttendEventCount uaec ON uaec.user_id = fap.user_id
          LEFT JOIN $tblEventCities ec ON ec.city = fap.event_city
          LEFT JOIN $tblEventCountries et ON et.country = fap.event_country
          LEFT JOIN $tblLocale l ON u.locale = l.locale
          LEFT JOIN $tblTimeZone tz ON tz.time_zone = u.time_zone"""
    //execute
    spark.sql(stmt)
  }

  //prepare feature set
  protected def prepareFeatureSet(dfUserFriendEvent: DataFrame): DataFrame = {
    //register
    dfUserFriendEvent.createOrReplaceTempView("user_friend_event")
    //register udf
    spark.udf.register("percentageValue", percentageValue)
    //statement
    val stmt ="""WITH
        min_max_train_data AS
        (
          SELECT
            MIN(user_age) AS user_min_age, MAX(user_age) AS user_max_age,
            MIN(user_member_days) AS user_min_member_days, MAX(user_member_days) AS user_max_member_days,
            MIN(user_friend_count) AS user_min_friend_count, MAX(user_friend_count) AS user_max_friend_count,
            MIN(user_invite_ahead_days) AS user_min_invite_ahead_days, MAX(user_invite_ahead_days) AS user_max_invite_ahead_days,
            MIN(user_had_event_count) AS user_min_had_event_count, MAX(user_had_event_count) AS user_max_had_event_count,
            MIN(user_invited_event_count) AS user_min_invited_event_count, MAX(user_invited_event_count) AS user_max_invited_event_count,
            MIN(user_attended_event_count) AS user_min_attended_event_count, MAX(user_attended_event_count) AS user_max_attended_event_count,
            MIN(user_not_attended_event_count) AS user_min_not_attended_event_count, MAX(user_not_attended_event_count) AS user_max_not_attended_event_count,
            MIN(user_maybe_attended_event_count) AS user_min_maybe_attended_event_count, MAX(user_maybe_attended_event_count) AS user_max_maybe_attended_event_count,
            MIN(user_invited_friends_count) AS user_min_invited_friends_count, MAX(user_invited_friends_count) AS user_max_invited_friends_count,
            MIN(user_attended_friends_count) AS user_min_attended_friends_count, MAX(user_attended_friends_count) AS user_max_attended_friends_count,
            MIN(user_not_attended_friends_count) AS user_min_not_attended_friends_count, MAX(user_not_attended_friends_count) AS user_max_not_attended_friends_count,
            MIN(user_maybe_attended_friends_count) AS user_min_maybe_attended_friends_count, MAX(user_maybe_attended_friends_count) AS user_max_maybe_attended_friends_count,
            MIN(event_start_ahead_days) AS event_min_start_ahead_days, MAX(event_start_ahead_days) AS event_max_start_ahead_days,
            MIN(event_invited_user_count) AS event_min_invited_user_count, MAX(event_invited_user_count) AS event_max_invited_user_count,
            MIN(event_attended_user_count) AS event_min_attended_user_count, MAX(event_attended_user_count) AS event_max_attended_user_count,
            MIN(event_not_attended_user_count) AS event_min_not_attended_user_count, MAX(event_not_attended_user_count) AS event_max_not_attended_user_count,
            MIN(event_maybe_attended_user_count) AS event_min_maybe_attended_user_count, MAX(event_maybe_attended_user_count) AS event_max_maybe_attended_user_count,
            MIN(latitude) AS event_min_latitude, MAX(latitude) AS event_max_latitude,
            MIN(longitude) AS event_min_longitude, MAX(longitude) AS event_max_longitude
          FROM user_friend_event
        ),
        pre_train_data AS
        (
          SELECT
            CASE WHEN m.user_interested IS NULL THEN 0 ELSE m.user_interested END AS user_interested,
            m.user_id,
            m.event_id,
            CASE WHEN m.user_locale IS NULL THEN 0 ELSE m.user_locale END AS user_locale,
            CASE WHEN m.user_gender IS NULL THEN -1 ELSE m.user_gender END AS user_gender,
            percentageValue(m.user_age, s.user_min_age, s.user_max_age) AS user_age,
            CASE WHEN m.user_time_zone IS NULL THEN 0 ELSE m.user_time_zone END AS user_time_zone,
            percentageValue(m.user_member_days, s.user_min_member_days, s.user_max_member_days) AS user_member_days,
            percentageValue(m.user_friend_count, s.user_min_friend_count, s.user_max_friend_count) AS user_friend_count,
            percentageValue(m.user_invite_ahead_days, s.user_min_invite_ahead_days, s.user_max_invite_ahead_days) AS user_invite_ahead_days,
            percentageValue(m.user_had_event_count, s.user_min_had_event_count, s.user_max_had_event_count) AS user_had_event_count,
            percentageValue(m.user_invited_event_count, s.user_min_invited_event_count, s.user_max_invited_event_count) AS user_invited_event_count,
            percentageValue(m.user_attended_event_count, s.user_min_attended_event_count, s.user_max_attended_event_count) AS user_attended_event_count,
            percentageValue(m.user_not_attended_event_count, s.user_min_not_attended_event_count, s.user_max_not_attended_event_count) AS user_not_attended_event_count,
            percentageValue(m.user_maybe_attended_event_count, s.user_min_maybe_attended_event_count, s.user_max_maybe_attended_event_count) AS user_maybe_attended_event_count,
            CASE WHEN m.user_invited IS NULL THEN 0 ELSE m.user_invited END AS user_invited,
            percentageValue(m.user_invited_friends_count, s.user_min_invited_friends_count, s.user_max_invited_friends_count) AS user_invited_friends_count,
            percentageValue(m.user_attended_friends_count, s.user_min_attended_friends_count, s.user_max_attended_friends_count) AS user_attended_friends_count,
            percentageValue(m.user_not_attended_friends_count, s.user_min_not_attended_friends_count, s.user_max_not_attended_friends_count) AS user_not_attended_friends_count,
            percentageValue(m.user_maybe_attended_friends_count, s.user_min_maybe_attended_friends_count, s.user_max_maybe_attended_friends_count) AS user_maybe_attended_friends_count,
            CASE WHEN m.user_invited_friends_percentage IS NULL THEN 0 ELSE m.user_invited_friends_percentage END AS user_invited_friends_percentage,
            CASE WHEN m.user_attended_friends_percentage IS NULL THEN 0 ELSE m.user_attended_friends_percentage END AS user_attended_friends_percentage,
            CASE WHEN m.user_not_attended_friends_percentage IS NULL THEN 0 ELSE m.user_not_attended_friends_percentage END AS user_not_attended_friends_percentage,
            CASE WHEN m.user_maybe_attended_friends_percentage IS NULL THEN 0 ELSE m.user_maybe_attended_friends_percentage END AS user_maybe_attended_friends_percentage,
            percentageValue(m.event_start_ahead_days, s.event_min_start_ahead_days, s.event_max_start_ahead_days) AS event_start_ahead_days,
            CASE WHEN m.event_start_month IS NULL THEN 0 ELSE m.event_start_month END AS event_start_month,
            CASE WHEN m.event_start_dayofweek IS NULL THEN 0 ELSE m.event_start_dayofweek END AS event_start_dayofweek,
            CASE WHEN m.event_start_hour IS NULL THEN 0 ELSE m.event_start_hour END AS event_start_hour,
            percentageValue(m.event_invited_user_count, s.event_min_invited_user_count, s.event_max_invited_user_count) AS event_invited_user_count,
            percentageValue(m.event_attended_user_count, s.event_min_attended_user_count, s.event_max_attended_user_count) AS event_attended_user_count,
            percentageValue(m.event_not_attended_user_count, s.event_min_not_attended_user_count, s.event_max_not_attended_user_count) AS event_not_attended_user_count,
            percentageValue(m.event_maybe_attended_user_count, s.event_min_maybe_attended_user_count, s.event_max_maybe_attended_user_count) AS event_maybe_attended_user_count,
            CASE WHEN m.event_city_level IS NULL THEN 0 ELSE m.event_city_level END AS event_city_level,
            CASE WHEN m.event_country_level IS NULL THEN 0 ELSE m.event_country_level END AS event_country_level,
            percentageValue(m.latitude, s.event_min_latitude, s.event_max_latitude) AS event_latitude,
            percentageValue(m.longitude, s.event_min_longitude, s.event_max_longitude) AS event_longitude,
            CASE WHEN m.event_creator_is_friend IS NULL THEN 0 ELSE m.event_creator_is_friend END AS event_creator_is_friend,
            CASE WHEN m.location_similar IS NULL THEN 0 ELSE m.location_similar END AS location_similar
          FROM user_friend_event m CROSS JOIN min_max_train_data s
        )
        SELECT
          CASE WHEN user_interested IS NULL THEN 0 ELSE user_interested END AS user_interested,
          user_id,
          event_id,
          CASE WHEN user_locale IS NULL THEN 0 ELSE user_locale END AS user_locale,
          CASE WHEN user_gender IS NULL THEN -1 ELSE user_gender END AS user_gender,
          CASE WHEN user_age IS NULL THEN 0.5 ELSE user_age END AS user_age,
          CASE WHEN user_time_zone IS NULL THEN 0 ELSE user_time_zone END AS user_time_zone,
          CASE WHEN user_member_days IS NULL THEN 0.5 ELSE user_member_days END AS user_member_days,
          CASE WHEN user_friend_count IS NULL THEN 0.5 ELSE user_friend_count END AS user_friend_count,
          CASE WHEN user_invite_ahead_days IS NULL THEN 0.5 ELSE user_invite_ahead_days END AS user_invite_ahead_days,
          CASE WHEN user_had_event_count IS NULL THEN 0.5 ELSE user_had_event_count END AS user_had_event_count,
          CASE WHEN user_invited_event_count IS NULL THEN 0.5 ELSE user_invited_event_count END AS user_invited_event_count,
          CASE WHEN user_attended_event_count IS NULL THEN 0.5 ELSE user_attended_event_count END AS user_attended_event_count,
          CASE WHEN user_not_attended_event_count IS NULL THEN 0.5 ELSE user_not_attended_event_count END AS user_not_attended_event_count,
          CASE WHEN user_maybe_attended_event_count IS NULL THEN 0.5 ELSE user_maybe_attended_event_count END AS user_maybe_attended_event_count,
          CASE WHEN user_invited IS NULL THEN 0 ELSE user_invited END AS user_invited,
          CASE WHEN user_invited_friends_count IS NULL THEN 0.5 ELSE user_invited_friends_count END AS user_invited_friends_count,
          CASE WHEN user_attended_friends_count IS NULL THEN 0.5 ELSE user_attended_friends_count END AS user_attended_friends_count,
          CASE WHEN user_not_attended_friends_count IS NULL THEN 0.5 ELSE user_not_attended_friends_count END AS user_not_attended_friends_count,
          CASE WHEN user_maybe_attended_friends_count IS NULL THEN 0.5 ELSE user_maybe_attended_friends_count END AS user_maybe_attended_friends_count,
          CASE WHEN user_invited_friends_percentage IS NULL THEN 0 ELSE user_invited_friends_percentage END AS user_invited_friends_percentage,
          CASE WHEN user_attended_friends_percentage IS NULL THEN 0 ELSE user_attended_friends_percentage END AS user_attended_friends_percentage,
          CASE WHEN user_not_attended_friends_percentage IS NULL THEN 0 ELSE user_not_attended_friends_percentage END AS user_not_attended_friends_percentage,
          CASE WHEN user_maybe_attended_friends_percentage IS NULL THEN 0 ELSE user_maybe_attended_friends_percentage END AS user_maybe_attended_friends_percentage,
          CASE WHEN event_start_ahead_days IS NULL THEN 0.5 ELSE event_start_ahead_days END AS event_start_ahead_days,
          CASE WHEN event_start_month IS NULL THEN 0 ELSE event_start_month END AS event_start_month,
          CASE WHEN event_start_dayofweek IS NULL THEN 0 ELSE event_start_dayofweek END AS event_start_dayofweek,
          CASE WHEN event_start_hour IS NULL THEN 0 ELSE event_start_hour END AS event_start_hour,
          CASE WHEN event_invited_user_count IS NULL THEN 0.5 ELSE event_invited_user_count END AS event_invited_user_count,
          CASE WHEN event_attended_user_count IS NULL THEN 0.5 ELSE event_attended_user_count END AS event_attended_user_count,
          CASE WHEN event_not_attended_user_count IS NULL THEN 0.5 ELSE event_not_attended_user_count END AS event_not_attended_user_count,
          CASE WHEN event_maybe_attended_user_count IS NULL THEN 0.5 ELSE event_maybe_attended_user_count END AS event_maybe_attended_user_count,
          CASE WHEN event_city_level IS NULL THEN 0 ELSE event_city_level END AS event_city_level,
          CASE WHEN event_country_level IS NULL THEN 0 ELSE event_country_level END AS event_country_level,
          CASE WHEN event_latitude IS NULL THEN 0.5 ELSE event_latitude END AS event_latitude,
          CASE WHEN event_longitude IS NULL THEN 0.5 ELSE event_longitude END AS event_longitude,
          CASE WHEN event_creator_is_friend IS NULL THEN 0 ELSE event_creator_is_friend END AS event_creator_is_friend,
          CASE WHEN location_similar IS NULL THEN 0 ELSE location_similar END AS location_similar
        FROM pre_train_data"""
    //execute
    spark.sql(stmt)
  }
}